package com.atguigu.bigdata.spark.core.acc;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.util.LongAccumulator;
import org.codehaus.janino.Java;

import java.util.Arrays;
/*
    累加器多加或者少加问题
        // 少加：转换算子中调用累加器，如果没有行动算子的话，那么不会执行
        // 多加：转换算子中调用累加器，如果没有行动算子的话，那么不会执行
        // 一般情况下，累加器会放置在行动算子进行操作

 */
public class Spark03_Acc_JAVA {
    public static void main(String[] args) {
        // 1.创建配置对象
        SparkConf conf = new SparkConf().setMaster("local[*]").setAppName("sparkCore");

        // 2. 创建sparkContext
        JavaSparkContext sc = new JavaSparkContext(conf);
        LongAccumulator AccSum = sc.sc().longAccumulator();

        JavaRDD<Integer> rdd = sc.parallelize(Arrays.asList(1,2,3,4,5));

        JavaRDD<Integer> map = rdd.map(new Function<Integer, Integer>() {
            @Override
            public Integer call(Integer v1) throws Exception {
                AccSum.add(v1);
                return v1;
            }
        });
        map.collect();
        map.collect();
        System.out.println(AccSum);

        sc.stop();
    }
}
